package studio.raptor.extractor.segment.task;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import studio.raptor.extractor.common.constant.EventMetaStatus;
import studio.raptor.extractor.common.holder.ExecutorHolder;
import studio.raptor.extractor.segment.config.ServerConfig;
import studio.raptor.extractor.segment.entity.EventMeta;
import studio.raptor.extractor.segment.service.EventMetaService;

/**
 * Created by liuyong8 on 2017/7/24.
 */
@Component
@Scope("prototype")
public class EventRecoverTask implements Runnable {

  private static final Logger log = LoggerFactory.getLogger(EventRecoverTask.class);

  @Autowired
  private EventMetaService eventMetaService;

  private ServerConfig serverConfig;

  @Override
  public void run() {
    // step2:event_meta表中剩余条数需要抽取
    boolean shouldContinue = true;
    List<String> statuses = new ArrayList<>();
    statuses.add(EventMetaStatus.STATUS_WAIT.value());
    statuses.add(EventMetaStatus.STATUS_PROCESS.value());
    while (shouldContinue) {
      List<EventMeta> eventMetas =
          eventMetaService.findMetasByKindIdAndStatuses(serverConfig.getKindId(), statuses);
      int leftHandleEventNum = eventMetas.size();
      if (leftHandleEventNum == 0) { // 停止
        serverConfig.setExtractorMQEnabled(new AtomicBoolean(true));
        shouldContinue = false;
        log.info("todo's event > left:0");
      } else {
        log.info("todo's event > left:" + leftHandleEventNum);
        LockSupport.parkUntil(System.currentTimeMillis() + 1000);
      }
    }
  }

  public void setServerConfig(ServerConfig serverConfig) {
    this.serverConfig = serverConfig;
  }
}
